package com.xix.canal.client.writer.mysql;

import com.alibaba.druid.pool.DruidDataSource;
import com.xix.canal.client.writer.mysql.config.ConfigLoader;
import com.xix.canal.client.writer.mysql.config.MysqlConfig;
import com.xix.canal.client.writer.mysql.config.MysqlDataSource;
import com.xix.canal.client.writer.mysql.monitor.MysqlConfigMonitor;
import com.xix.canal.client.writer.mysql.service.MysqlSyncService;
import com.xixi.canal.client.DataWriter;
import com.xixi.canal.client.support.Dml;
import com.xixi.canal.client.support.SPI;
import com.xixi.canal.client.support.WriterConfig;
import com.xixi.canal.client.util.Assert;
import com.xixi.canal.client.util.StringUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;

@SPI("mysql")
public class MysqlWriter implements DataWriter {

    private static Logger logger = LoggerFactory.getLogger(MysqlWriter.class);


    private Properties properties;

    private String destination;

    private WriterConfig writerConfig;

    // filename - config
    private Map<String, MysqlConfig> mysqlConfigMap = new ConcurrentHashMap<>();

    // dataKey - config
    // destination + database + table
    private Map<String, Map<String, MysqlConfig>> dbMysqlConfigMap = new ConcurrentHashMap<>();

    // destination + database + table -> destination + database
    private Map<String, MysqlDataSource> dbConfigMap = new ConcurrentHashMap<>();

    // destination + database
    private Map<String, DruidDataSource> dataSourceMap = new ConcurrentHashMap<>();

    private MysqlSyncService mysqlSyncService;

    private MysqlConfigMonitor mysqlConfigMonitor;

    @Override
    public void init(String destination, WriterConfig writerConfig, Properties envProperties) {
        this.properties = properties;
        this.destination = destination;
        this.writerConfig = writerConfig;

        mysqlConfigMap = ConfigLoader.load(writerConfig, destination, envProperties);

        for (Map.Entry<String, MysqlConfig> entry : mysqlConfigMap.entrySet()) {
            MysqlConfig mysqlConfig = entry.getValue();

            String dataSourceKey = StringUtil.getDataSourceKey(destination,
                    mysqlConfig.getTarget().getSourceDB());
            dataSourceMap.computeIfAbsent(dataSourceKey,
                    v -> buildDruidDataSource(mysqlConfig));

            String key = StringUtil.getDataSourceKey(destination,
                    mysqlConfig.getTarget().getSourceTable());
            Map<String, MysqlConfig> stringMysqlConfigMap = dbMysqlConfigMap.computeIfAbsent(key, k -> new ConcurrentHashMap<>());
            stringMysqlConfigMap.putIfAbsent(entry.getKey(), mysqlConfig);
        }

        mysqlSyncService = new MysqlSyncService(dataSourceMap);

        mysqlConfigMonitor = new MysqlConfigMonitor(this, properties,
                destination, writerConfig);
        mysqlConfigMonitor.init();
    }

    private DruidDataSource buildDruidDataSource(MysqlConfig mysqlConfig) {
        MysqlConfig.DataSourceInfo dataSourceInfo = mysqlConfig.getDataSource();
        DruidDataSource dataSource = new DruidDataSource();
        Assert.notNull(dataSourceInfo, "Datasource cannot empty.");
        dataSource = new DruidDataSource();
        dataSource.setDriverClassName(dataSourceInfo.getDriverClassName());
        dataSource.setUrl(dataSourceInfo.getUrl());
        dataSource.setUsername(dataSourceInfo.getUsername());
        dataSource.setPassword(dataSourceInfo.getPassword());
        dataSource.setInitialSize(1);
        dataSource.setMinIdle(1);
        dataSource.setMaxActive(30);
        dataSource.setMaxWait(60000);
        dataSource.setTimeBetweenEvictionRunsMillis(60000);
        dataSource.setMinEvictableIdleTimeMillis(300000);
        dataSource.setUseUnfairLock(true);
        try {
            dataSource.init();
        } catch (SQLException e) {
            logger.error("ERROR ## failed to initial datasource: " + properties.get("jdbc.url"), e);
        }
        return dataSource;
    }


    @Override
    public void write(List<Dml> list) {

        if (null == list || list.isEmpty()) return;

        try {
            mysqlSyncService.sync(dbMysqlConfigMap, list, properties);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    @Override
    public void destroy() {

        if (null != mysqlSyncService) {
            mysqlSyncService.close();
        }

        if (dataSourceMap != null && !dataSourceMap.isEmpty()) {
            for (Map.Entry<String, DruidDataSource> entry : dataSourceMap.entrySet()) {
                entry.getValue().close();
            }
            dataSourceMap.clear();
        }

        if (null != mysqlConfigMap && !mysqlConfigMap.isEmpty()) {
            mysqlConfigMap.clear();
        }
        if (null != dbConfigMap && !dbConfigMap.isEmpty()) {
            dbConfigMap.clear();
        }

        if (null != dbMysqlConfigMap && !dbMysqlConfigMap.isEmpty()) {
            dbMysqlConfigMap.clear();
        }

        if (null != mysqlConfigMonitor) {
            mysqlConfigMonitor.destroy();
        }
    }


}
